package org.tao.lightningmq.nameserver.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import org.tao.lightningmq.common.dto.PullBrokerIpRespDTO;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.enums.RegistryTypeEnum;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.enums.PullBrokerIpRoleEnum;
import org.tao.lightningmq.nameserver.event.model.PullBrokerIpEvent;
import org.tao.lightningmq.nameserver.store.ServiceInstance;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author lht
 * @date 2025/1/16 - 21:46
 * @description:
 */
public class PullBrokerIpListener implements Listener<PullBrokerIpEvent> {
    @Override
    public void onReceive(PullBrokerIpEvent event) throws Exception {
        String pullRole = event.getRole();
        Map<String, ServiceInstance> serviceInstanceMap = CommonCache.getServiceInstanceManager().getServiceInstanceMap();
        PullBrokerIpRespDTO pullBrokerIpRespDTO = new PullBrokerIpRespDTO();
        List<String> addressList = new ArrayList<>();
        List<String> masterAddressList = new ArrayList<>();
        List<String> slaveAddressList = new ArrayList<>();
        for (String reqId : serviceInstanceMap.keySet()) {
            ServiceInstance serviceInstance = serviceInstanceMap.get(reqId);
            if (!RegistryTypeEnum.BROKER.getCode().equals(serviceInstance.getRegistryType())) {
                continue;
            }
            Map<String,Object> brokerAttrs = serviceInstance.getAttrs();
            String group = (String) brokerAttrs.getOrDefault("group","");
            //根据角色进行判断
            String role = (String) brokerAttrs.get("role");
            if(group.equals(event.getBrokerClusterGroup())) {
                if (
//                        PullBrokerIpRoleEnum.MASTER.getCode().equals(pullRole) &&
                        PullBrokerIpRoleEnum.MASTER.getCode().equals(role)) {
                    masterAddressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                } else if (
//                        PullBrokerIpRoleEnum.SLAVE.getCode().equals(pullRole) &&
                        PullBrokerIpRoleEnum.SLAVE.getCode().equals(role)) {
                    slaveAddressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                } else if (
//                        PullBrokerIpRoleEnum.SINGLE.getCode().equals(pullRole)&&
                        PullBrokerIpRoleEnum.SINGLE.getCode().equals(role)) {
                    addressList.add(serviceInstance.getIp()+":"+serviceInstance.getPort());
                }
            }
        }
        pullBrokerIpRespDTO.setMsgId(event.getMsgId());
        pullBrokerIpRespDTO.setMasterAddressList(masterAddressList);
        pullBrokerIpRespDTO.setSlaveAddressList(slaveAddressList);
        //防止ip重复
        pullBrokerIpRespDTO.setAddressList(addressList.stream().distinct().collect(Collectors.toList()));
        event.getContext().writeAndFlush(new TcpMsg(NameServerResponseCode.PULL_BROKER_ADDRESS_SUCCESS.getCode(),
                JSON.toJSONBytes(pullBrokerIpRespDTO)));
    }
}
